package com.yujiahao.bigdata.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.text.SimpleDateFormat
import java.util.Date

//TODO 需求二：广告点击量实时统计--描述：实时统计每天各地区各城市各广告的点击总流量，并将其存入MySQL。
object Stream_Source_Test_Kafka2_1 {
  def main(args: Array[String]): Unit = {

    //TODO SparkStreaming环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))


    //TODO 4、业务逻辑-- 从kafka中采集数据
    //在流式数据处理中，kafka应用特别多，所以很多的框架都有相应的工具类

    //4.1.定义Kafka参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "atguigu", //消费者组名
      //KV反序列化
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //4.2、kafka消费参数
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String]( //这个是读取Kafka的KV类型
        ssc, //上面的环境对象
        LocationStrategies.PreferConsistent, //位置策略，由框架自行选择
        //消费策略
        ConsumerStrategies.Subscribe[String, String](Set("atguigu210819"), kafkaPara))

    //因为kafka中K和分区策略有关系，因此这里数据直接就Values
    val kafkaValuse: DStream[String] = kafkaDStream.map(_.value())
    //TODO 处理kafka的数据
    //先将数据转换一下结构然后对每一条rdd进行WordCount
    val wordCount: DStream[((String, String, String, String), Long)] = kafkaValuse.map(
      line => {

        val datas: Array[String] = line.split(" ")
        //因为要求是统计每一天的时间，这里的是毫秒，所以要将时间转换一下格式
        val times = new SimpleDateFormat("yy-MM-dd")
        //将数据中的时间传进去进行处理
        val newTime: String = times.format(new Date(datas(0).toLong))
        //Map中转换的结构应该是下面的一个状况,然后在进行聚合求出每个时间的广告总点击量
        ((newTime, datas(1), datas(2), datas(4)), 1L)
      }
    ).reduceByKey(_ + _)

    //TODO 接下来将数据更新到MySQL中去，可以使用DataFrame，也可以直接使用JDBC 传递，这里使用JDBC比价简单的
   //这里没有提供直接写在MySQL中的API所以就的自己手动将RDD中的数据放进去
      wordCount.foreachRDD(
        rdd =>{
          //这里可以解决频繁连接mysql以及分布式数据大量连接一个MySQL从而导致性能底下的问题

          rdd.foreachPartition(
            iter => {
            //优化采用连接池实现mysql的连接
              val conn: Connection = JdbcUtil.getConnection
              // 将RDD中的数据遍历出来保存到MySQL中去
              iter.foreach{
                //这里直接用模式匹配，因为知道数据的格式了就直接处理了
                case ((day, area,city,ad),sum) =>{

                  // 连接之后那么就可以开始操作了,直接插入的时候可能会有主键冲突，因为消费的数据可能会重复，所以加一个命令
                  val pstat: PreparedStatement = conn.prepareStatement(
                    """
                      |insert into area_city_ad_count
                      |(
                      |dt,
                      |area,
                      |city,
                      |adid,
                      |count
                      |)
                      |values(
                      |?,?,?,?,?
                      |)
                      | ON DUPLICATE KEY
                      | update count = count+?
                      |""".stripMargin
                  )
                  //接下来要传入参数，参数是从1 开始的
                  pstat.setString(1, day)
                  pstat.setString(2, area)
                  pstat.setString(3, city)
                  pstat.setString(4, ad)
                  pstat.setLong(5, sum)
                  //这里就是处理主键冲突的
                  pstat.setLong(6, sum)
                  pstat.executeUpdate()

                  //关闭资源
                  pstat.close()

                }
              }
              conn.close()
            }
          )


        }

      )


    //TODO 2、启动采集器
    ssc.start()
    //TODO 3、Driver等待采集器的结束，否则，当前Driver处于阻塞状态
    ssc.awaitTermination()

  }
}
